package cn.trigram.concurrent.pool;

import cn.trigram.concurrent.ThreadPoolUtil;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.AbortPolicy;
import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.Validate;

/**
 * 使用唯品会的代码 ThreadPool创建的工具类.
 * <p>
 * 对比JDK Executors中的newFixedThreadPool(), newCachedThreadPool(),newScheduledThreadPool, 提供更多有用的配置项.
 * <p>
 * 另包含了移植自Tomcat的QueuableCachedPool.
 * <p>
 * 使用示例如下：
 *
 * <pre>
 * ExecutorService ExecutorService = new FixedThreadPoolBuilder().setPoolSize(10).build();
 * </pre>
 * <p>
 * 参考文章 《Java ThreadPool的正确打开方式》http://calvin1978.blogcn.com/articles/java-threadpool.html
 */
public class ThreadPoolBuilder {

  private static final RejectedExecutionHandler defaultRejectHandler = new AbortPolicy();

  /**
   * 固定线程池构建
   *
   * @see FixedThreadPoolBuilder
   */
  public static FixedThreadPoolBuilder fixedPool() {

    return new FixedThreadPoolBuilder();
  }

  /**
   * 可缓存线程的线程池，线程空闲10秒才销毁 更建议使用{@link #queuableCachedPool()}
   *
   * @see CachedThreadPoolBuilder
   */
  public static CachedThreadPoolBuilder cachedPool() {

    return new CachedThreadPoolBuilder();
  }

  /**
   * @see ScheduledThreadPoolBuilder
   */
  public static ScheduledThreadPoolBuilder scheduledPool() {

    return new ScheduledThreadPoolBuilder();
  }

  /**
   * 可缓存线程的同时，具有任务队列的线程池
   *
   * @see QueuableCachedThreadPoolBuilder
   */
  public static QueuableCachedThreadPoolBuilder queuableCachedPool() {

    return new QueuableCachedThreadPoolBuilder();
  }

  /**
   * 基于权重的线程池
   *
   * @see WeightThreadPoolBuilder
   */
  public static WeightThreadPoolBuilder weightThreadPool() {

    return new WeightThreadPoolBuilder();
  }

  /**
   * 优先使用threadFactory，否则如果threadNamePrefix不为空则使用自建ThreadFactory，否则使用defaultThreadFactory
   */
  private static ThreadFactory createThreadFactory(
      ThreadFactory threadFactory,
      String threadNamePrefix,
      Boolean daemon
  ) {

    if (threadFactory != null) {
      return threadFactory;
    }

    if (threadNamePrefix != null) {
      if (daemon != null) {
        return ThreadPoolUtil.buildThreadFactory(threadNamePrefix, daemon);
      } else {
        return ThreadPoolUtil.buildThreadFactory(threadNamePrefix);
      }
    }

    return Executors.defaultThreadFactory();
  }

  /**
   * 基于权重的线程池构建器
   *
   * @version 1.0.0
   * @since 1.0.0
   */
  public static class WeightThreadPoolBuilder {

    /**
     * 线程池最小线程数
     */
    private int minSize = 0;

    /**
     * 线程池的最大线程数
     */
    private int maxSize = Integer.MAX_VALUE;

    /**
     * 队列的容量
     */
    private int queueSize = Integer.MAX_VALUE;

    /**
     * 空闲线程存留时间，毫秒单位
     */
    private int   keepAliveMillis        = 10_000;

    /**
     * 队列保留容量与总容量的比例
     */
    private float reservedQueueSizeRatio = 0f;

    /**
     * 可进入队列保留容量的权重阈值
     */
    private int weightThreshold = Integer.MIN_VALUE;

    private ThreadFactory threadFactory;

    /**
     * 线程名称前缀
     */
    private String threadNamePrefix;

    /**
     * 创建的线程是否是守护/后台线程
     */
    private Boolean daemon;


    public WeightThreadPoolBuilder setMinSize(int minSize) {

      this.minSize = minSize;
      return this;
    }

    /**
     * Max默认Integer.MAX_VALUE的，建议设置
     */
    public WeightThreadPoolBuilder setMaxSize(int maxSize) {

      this.maxSize = maxSize;
      return this;
    }

    /**
     * 默认使用{@link Integer#MAX_VALUE}
     */
    public WeightThreadPoolBuilder setQueueSize(int queueSize) {

      this.queueSize = queueSize;
      return this;
    }

    /**
     * JDK默认值60秒太高，如高达1000线程时，要低于16QPS时才会开始回收线程, 因此改为默认10秒.
     */
    public WeightThreadPoolBuilder setKeepAliveMillis(int keepAliveMillis) {

      this.keepAliveMillis = keepAliveMillis;
      return this;
    }

    /**
     * 默认0
     */
    public WeightThreadPoolBuilder setReservedQueueSizeRatio(float reservedQueueSizeRatio) {

      this.reservedQueueSizeRatio = reservedQueueSizeRatio;
      return this;
    }

    /**
     * 与threadNamePrefix互斥, 优先使用ThreadFactory
     */
    public WeightThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {

      this.threadFactory = threadFactory;
      return this;
    }

    /**
     * 与ThreadFactory互斥, 优先使用ThreadFactory
     */
    public WeightThreadPoolBuilder setThreadNamePrefix(String threadNamePrefix) {

      this.threadNamePrefix = threadNamePrefix;
      return this;
    }

    /**
     * 与threadFactory互斥, 优先使用ThreadFactory
     * <p>
     * 默认为NULL，不进行设置，使用JDK的默认值.
     */
    public WeightThreadPoolBuilder setDaemon(Boolean daemon) {

      this.daemon = daemon;
      return this;
    }

    public WeightThreadPoolBuilder setWeightThreshold(int weightThreshold) {

      this.weightThreshold = weightThreshold;
      return this;
    }

    public ThreadPoolExecutor build() {

      WeightThreadPoolExecutor.WeightQueue queue = new WeightThreadPoolExecutor.WeightQueue(
          queueSize, reservedQueueSizeRatio);

      threadFactory = createThreadFactory(threadFactory, threadNamePrefix, daemon);

      RejectedExecutionHandler rejectHandler = new WeightThreadPoolExecutor.WeightPolicyRejectedExecutionHandler(
          this.weightThreshold);

      return new WeightThreadPoolExecutor(
          minSize, maxSize,
          keepAliveMillis, TimeUnit.MILLISECONDS,
          queue,
          threadFactory,
          rejectHandler
      );
    }

  }

  /**
   * 创建FixedThreadPool.建议必须设置queueSize保证有界。
   * <p>
   * 1. 任务提交时, 如果线程数还没达到poolSize即创建新线程并绑定任务(即poolSize次提交后线程总数必达到poolSize，不会重用之前的线程)
   * <p>
   * poolSize默认为1，即singleThreadPool.
   * <p>
   * 2. 第poolSize次任务提交后, 新增任务放入Queue中, Pool中的所有线程从Queue中take任务执行.
   * <p>
   * Queue默认为无限长的LinkedBlockingQueue, 但建议设置queueSize换成有界的队列.
   * <p>
   * 如果使用有界队列, 当队列满了之后,会调用RejectHandler进行处理, 默认为AbortPolicy，抛出RejectedExecutionException异常.
   * 其他可选的Policy包括静默放弃当前任务(Discard)，放弃Queue里最老的任务(DisacardOldest)，或由主线程来直接执行(CallerRuns).
   * <p>
   * 3. 因为线程全部为core线程，所以不会在空闲时回收.
   */
  public static class FixedThreadPoolBuilder {

    private int poolSize = 1;

    private int queueSize = -1;

    private ThreadFactory threadFactory;

    private String threadNamePrefix;

    private Boolean daemon;

    private RejectedExecutionHandler rejectHandler;

    /**
     * Pool大小，默认为1，即singleThreadPool
     */
    public FixedThreadPoolBuilder setPoolSize(int poolSize) {

      Validate.isTrue(poolSize >= 1);
      this.poolSize = poolSize;
      return this;
    }

    /**
     * 不设置时默认为-1, 使用无限长的LinkedBlockingQueue.
     * <p>
     * 为正数时使用ArrayBlockingQueue.
     */
    public FixedThreadPoolBuilder setQueueSize(int queueSize) {

      this.queueSize = queueSize;
      return this;
    }

    /**
     * 与threadNamePrefix互斥, 优先使用ThreadFactory
     */
    public FixedThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {

      this.threadFactory = threadFactory;
      return this;
    }

    /**
     * 与ThreadFactory互斥, 优先使用ThreadFactory
     */
    public FixedThreadPoolBuilder setThreadNamePrefix(String threadNamePrefix) {

      this.threadNamePrefix = threadNamePrefix;
      return this;
    }

    /**
     * 与threadFactory互斥, 优先使用ThreadFactory
     * <p>
     * 默认为NULL，不进行设置，使用JDK的默认值.
     */
    public FixedThreadPoolBuilder setDaemon(Boolean daemon) {

      this.daemon = daemon;
      return this;
    }

    public FixedThreadPoolBuilder setRejectHanlder(RejectedExecutionHandler rejectHandler) {

      this.rejectHandler = rejectHandler;
      return this;
    }

    public ThreadPoolExecutor build() {

      BlockingQueue<Runnable> queue = null;
      if (queueSize < 1) {
        queue = new LinkedBlockingQueue<Runnable>();
      } else {
        queue = new ArrayBlockingQueue<Runnable>(queueSize);
      }

      threadFactory = createThreadFactory(threadFactory, threadNamePrefix, daemon);

      if (rejectHandler == null) {
        rejectHandler = defaultRejectHandler;
      }

      return new ThreadPoolExecutor(poolSize, poolSize, 0L, TimeUnit.MILLISECONDS, queue,
          threadFactory,
          rejectHandler
      );
    }

  }

  /**
   * <b>适用场景：必须处理任务，不允许任务空置；原理，只要有任务，线程不够就创建线程，直到maxSize上限。</b>
   * 更建议使用{@link QueuableCachedThreadPoolBuilder} <br/> 创建CachedThreadPool, maxSize建议设置
   * <p>
   * 1. 任务提交时, 如果线程数还没达到minSize即创建新线程并绑定任务(即minSize次提交后线程总数必达到minSize, 不会重用之前的线程)
   * <p>
   * minSize默认为0, 可设置保证有基本的线程处理请求不被回收.
   * <p>
   * 2. 第minSize次任务提交后, 新增任务提交进SynchronousQueue后，如果没有空闲线程立刻处理，则会创建新的线程, 直到总线程数达到上限.
   * <p>
   * maxSize默认为Integer.Max, 可以进行设置.
   * <p>
   * 如果设置了maxSize, 当总线程数达到上限, 会调用RejectHandler进行处理, 默认为AbortPolicy, 抛出RejectedExecutionException异常.
   * 其他可选的Policy包括静默放弃当前任务(Discard)，或由主线程来直接执行(CallerRuns).
   * <p>
   * 3. minSize以上, maxSize以下的线程, 如果在keepAliveTime中都poll不到任务执行将会被结束掉, keeAliveTimeJDK默认为10秒.
   * JDK默认值60秒太高，如高达1000线程时，要低于16QPS时才会开始回收线程, 因此改为默认10秒.
   */
  public static class CachedThreadPoolBuilder {

    private int minSize = 0;

    private int maxSize = Integer.MAX_VALUE;

    private int keepAliveSecs = 10;

    private ThreadFactory threadFactory;

    private String threadNamePrefix;

    private Boolean daemon;

    private RejectedExecutionHandler rejectHandler;

    public CachedThreadPoolBuilder setMinSize(int minSize) {

      this.minSize = minSize;
      return this;
    }

    /**
     * Max默认Integer.MAX_VALUE的，建议设置
     */
    public CachedThreadPoolBuilder setMaxSize(int maxSize) {

      this.maxSize = maxSize;
      return this;
    }

    /**
     * JDK默认值60秒太高，如高达1000线程时，要低于16QPS时才会开始回收线程, 因此改为默认10秒.
     */
    public CachedThreadPoolBuilder setKeepAliveSecs(int keepAliveSecs) {

      this.keepAliveSecs = keepAliveSecs;
      return this;
    }

    /**
     * 与threadNamePrefix互斥, 优先使用ThreadFactory
     */
    public CachedThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {

      this.threadFactory = threadFactory;
      return this;
    }

    /**
     * 与threadFactory互斥, 优先使用ThreadFactory
     */
    public CachedThreadPoolBuilder setThreadNamePrefix(String threadNamePrefix) {

      this.threadNamePrefix = threadNamePrefix;
      return this;
    }

    /**
     * 与threadFactory互斥, 优先使用ThreadFactory
     * <p>
     * 默认为NULL，不进行设置，使用JDK的默认值.
     */
    public CachedThreadPoolBuilder setDaemon(Boolean daemon) {

      this.daemon = daemon;
      return this;
    }

    public CachedThreadPoolBuilder setRejectHanlder(RejectedExecutionHandler rejectHandler) {

      this.rejectHandler = rejectHandler;
      return this;
    }

    public ThreadPoolExecutor build() {

      threadFactory = createThreadFactory(threadFactory, threadNamePrefix, daemon);

      if (rejectHandler == null) {
        rejectHandler = defaultRejectHandler;
      }

      return new ThreadPoolExecutor(minSize, maxSize, keepAliveSecs, TimeUnit.SECONDS,
          new SynchronousQueue<Runnable>(), threadFactory, rejectHandler
      );
    }

  }

  /**
   * 创建ScheduledPool.
   */
  public static class ScheduledThreadPoolBuilder {

    private int poolSize = 1;

    private ThreadFactory threadFactory;

    private String threadNamePrefix;

    /**
     * 默认为1
     */
    public ScheduledThreadPoolBuilder setPoolSize(int poolSize) {

      this.poolSize = poolSize;
      return this;
    }

    /**
     * 与threadNamePrefix互斥, 优先使用ThreadFactory
     */
    public ScheduledThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {

      this.threadFactory = threadFactory;
      return this;
    }

    public ScheduledThreadPoolBuilder setThreadNamePrefix(String threadNamePrefix) {

      this.threadNamePrefix = threadNamePrefix;
      return this;
    }

    public ScheduledThreadPoolExecutor build() {

      threadFactory = createThreadFactory(threadFactory, threadNamePrefix, Boolean.TRUE);
      return new ScheduledThreadPoolExecutor(poolSize, threadFactory);
    }

  }

  /**
   * 从Tomcat移植过来的可扩展可用Queue缓存任务的ThreadPool <br/> 与CachedThreadPoolBuilder不同的是，同时提供了一个任务队列，当到达maxSize上限后，任务会继续缓存在队列中
   *
   * @see QueuableCachedThreadPoolExecutor
   */
  public static class QueuableCachedThreadPoolBuilder {

    private int minSize = 0;

    private int maxSize = Integer.MAX_VALUE;

    private int keepAliveSecs = 10;

    private int queueSize = 100;

    private ThreadFactory threadFactory;

    private String threadNamePrefix;

    private Boolean daemon;

    private RejectedExecutionHandler rejectHandler;

    public QueuableCachedThreadPoolBuilder setMinSize(int minSize) {

      this.minSize = minSize;
      return this;
    }

    public QueuableCachedThreadPoolBuilder setMaxSize(int maxSize) {

      this.maxSize = maxSize;
      return this;
    }

    /**
     * LinkedQueue长度, 默认100
     */
    public QueuableCachedThreadPoolBuilder setQueueSize(int queueSize) {

      this.queueSize = queueSize;
      return this;
    }

    public QueuableCachedThreadPoolBuilder setKeepAliveSecs(int keepAliveSecs) {

      this.keepAliveSecs = keepAliveSecs;
      return this;
    }

    /**
     * 与threadNamePrefix互斥, 优先使用ThreadFactory
     */
    public QueuableCachedThreadPoolBuilder setThreadFactory(ThreadFactory threadFactory) {

      this.threadFactory = threadFactory;
      return this;
    }

    /**
     * 与threadFactory互斥, 优先使用ThreadFactory
     */
    public QueuableCachedThreadPoolBuilder setThreadNamePrefix(String threadNamePrefix) {

      this.threadNamePrefix = threadNamePrefix;
      return this;
    }

    /**
     * 与threadFactory互斥, 优先使用ThreadFactory
     * <p>
     * 默认为NULL，不进行设置，使用JDK的默认值.
     */
    public QueuableCachedThreadPoolBuilder setDaemon(Boolean daemon) {

      this.daemon = daemon;
      return this;
    }

    public QueuableCachedThreadPoolBuilder setRejectHanlder(
        RejectedExecutionHandler rejectHandler
    ) {

      this.rejectHandler = rejectHandler;
      return this;
    }

    public QueuableCachedThreadPoolExecutor build() {

      threadFactory = createThreadFactory(threadFactory, threadNamePrefix, daemon);

      if (rejectHandler == null) {
        rejectHandler = defaultRejectHandler;
      }

      return new QueuableCachedThreadPoolExecutor(minSize, maxSize, keepAliveSecs, TimeUnit.SECONDS,
          new QueuableCachedThreadPoolExecutor.ControllableQueue(queueSize), threadFactory, rejectHandler
      );
    }

  }

}
